package org.apache.flink.connector.httptm;

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.formats.jsontm.JsonTmDecodingFormat;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.factories.DeserializationFormatFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.types.DataType;

import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * <ol>
 * date:2022/4/6 editor:yanghongjian
 * <li>创建文档-外部接口工厂 添加到 META-INF/services 目录</li>
 * <li>工厂的目的是验证 options、配置编码/解码格式（如果需要）、创建 Table connector 的参数化实例</li>
 * <li>DataApiDynamicTableSourceFactory 将 CatalogTable 转换为 TableSource实例</li>
 * <li>添加到 META-INF/services 目录</li>
 * SET execution.type = batch;
 * SET execution.result-mode = tableau;
 * <pre>
 * CREATE TABLE IF NOT EXISTS dataapiout_table (
 * `msg` STRING COMMENT '',
 * `total` DECIMAL(32,0) COMMENT ''
 * )COMMENT '定义外部接口表'
 * WITH(
 * 'connector' = 'dataapiOut',
 * 'format' = 'json-wide',
 * 'url' = 'http://localhost/api/test/parames',
 * 'method' = 'GET',
 * 'setting' = '{"timeout":0}',
 * 'queryParams' = '',
 * 'authorization' = '{"type":"","value":"","value1":"","value2":"","addTo":""}',
 * 'headers' = '[{"id":"202201051038038","key":"Content-Type","keyEdit":false,"value":"application/json; charset=utf-8","valueEdit":false,"comm":"","commEdit":false,"isSel":true}]',
 * 'body' = '{"type":"","value":""}'
 * );
 *
 * SELECT
 * `msg`, `total`
 * FROM dataapiout_table;
 * </pre>
 * </ol>
 * <ol>
 *
 * </ol>
 * <ol>
 *
 * @author <a href="mailto:12719889@qq.com">YangHongJian</a>
 * @version 3.0
 * @since 1.8
 */
public class HttpTmTableSourceFactory implements DynamicTableSourceFactory {
    // 定义 options
    public static final ConfigOption<String> URL = ConfigOptions.key("url")
            .stringType().noDefaultValue();
    //默认连接方式：GET
    public static final ConfigOption<String> METHOD = ConfigOptions.key("method")
            .stringType().defaultValue("GET");
    public static final ConfigOption<String> HEADERS = ConfigOptions.key("headers")
            .stringType().defaultValue("{}");
    public static final ConfigOption<String> QUERYPARAMS = ConfigOptions.key("queryParams")
            .stringType().defaultValue("{}");
    public static final ConfigOption<String> BODY = ConfigOptions.key("body")
            .stringType().defaultValue("");
    public static final ConfigOption<String> SETTING = ConfigOptions.key("setting")
            .stringType().defaultValue("{}");

    // 工厂标识符，用来匹配 'connector'='http-tm'
    @Override
    public String factoryIdentifier() {
        return "http-tm";
    }

    // 定义必填 options
    @Override
    public Set<ConfigOption<?>> requiredOptions() {
        final Set<ConfigOption<?>> options = new HashSet<>();
        options.add(URL);
        options.add(FactoryUtil.FORMAT);
        return options;
    }

    // 定义可选 options
    @Override
    public Set<ConfigOption<?>> optionalOptions() {
        final Set<ConfigOption<?>> options = new HashSet<>();
        options.add(METHOD);
        options.add(HEADERS);
        options.add(QUERYPARAMS);
        options.add(BODY);
        options.add(SETTING);
        return options;
    }

    @Override
    public DynamicTableSource createDynamicTableSource(Context context) {
        // 实现自定义验证逻辑 或 使用提供的帮助工具类 FactoryUtil.TableFactoryHelper
        final FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
        // 发现合适的 DecodingFormat
        final DecodingFormat<DeserializationSchema<RowData>> decodingFormat = helper.discoverDecodingFormat(
                DeserializationFormatFactory.class, FactoryUtil.FORMAT);

        // 验证 options
        helper.validate();

        final ResolvedSchema resolvedSchema = context.getCatalogTable().getResolvedSchema();
        // 从 CatalogTable 派生生成的数据类型（不包括计算列）
        final DataType physicalRowDataType = resolvedSchema.toPhysicalRowDataType();
        //获取metadata列
        final List<Column.MetadataColumn> metadataColumns = resolvedSchema.getColumns().stream()
                .filter(Column.MetadataColumn.class::isInstance)
                .map(Column.MetadataColumn.class::cast)
                .collect(Collectors.toList());

        if (decodingFormat instanceof JsonTmDecodingFormat) {
            ((JsonTmDecodingFormat) decodingFormat).addReadableMetadata(metadataColumns);
        }
        // 创建并返回 DynamicTableSource
//        Map<String, String> mapOpt = ((Configuration) options).toMap();
        Map<String, String> mapCntOpt = ((Configuration) context.getConfiguration()).toMap();
        return new HttpTmTableSource(physicalRowDataType, metadataColumns, helper.getOptions(), decodingFormat,
                //批模式时为有界数据
                "BATCH".equals(mapCntOpt.get("execution.runtime-mode")));
    }
}
